package com.example.raft;

import lombok.Data;
import org.jgroups.*;
import org.jgroups.protocols.raft.RAFT;
import org.jgroups.protocols.raft.Role;
import org.jgroups.raft.RaftHandle;
import org.jgroups.raft.blocks.ReplicatedStateMachine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.UUID;

/**
 * @className: RaftServer
 * @author: chentao
 * @date: 2025/9/18 8:20
 * @Version: 1.0
 * @description:
 */
@Data
public class RaftServer {

    static Logger logger = LoggerFactory.getLogger(RaftServer.class.getName());

    public static final String DEFAULT_SERVER = "DEFAULT";

    private String serverId;

    private View view;

    private JChannel jChannel;

    private RaftHandle raftHandle;

    private RaftListener raftListener = new RaftListener() {
        @Override
        public void memberJoin(List<Address> address) {
            RaftListener.super.memberJoin(address);
        }

        @Override
        public void memberLeave(List<Address> address) {
            RaftListener.super.memberLeave(address);
        }

        @Override
        public void receive(Message message) {
            RaftListener.super.receive(message);
        }

        @Override
        public void isLeader() {
            RaftListener.super.isLeader();
        }

        @Override
        public void isNotLeader() {
            RaftListener.super.isNotLeader();
        }
    };

    private ReplicatedStateMachine replicatedStateMachine;

    public RaftServer() {
        this.serverId = UUID.randomUUID().toString();
    }

    public RaftServer(String serverId) {
        this.serverId = serverId;
    }

    public RaftServer(String serverId, RaftListener raftListener) {
        this.serverId = serverId;
        this.raftListener = raftListener;
    }

    /***
     * 启动raft server
     *
     * @throws Exception
     */
    public void start() throws Exception {
        this.jChannel = new JChannel("jgroups-udp.xml").name(serverId);
        this.replicatedStateMachine = new ReplicatedStateMachine(this.jChannel);
        this.raftHandle = new RaftHandle(this.jChannel, this.replicatedStateMachine).raftId(serverId);
        this.raftHandle.raft().setMembers(this.serverId);
        this.raftHandle.addRoleListener(new RAFT.RoleChange() {
            @Override
            public void roleChanged(Role role) {
                if (role == Role.Leader) {
                    raftListener.isLeader();
                } else {
                    raftListener.isNotLeader();
                }
            }
        });
        this.jChannel.setReceiver(new ReceiverAdapter() {
            @Override
            public void receive(Message msg) {
                raftListener.receive(msg);
            }

            @Override
            public void viewAccepted(View new_view) {
                if (view != null) {
                    List<Address> leftMembers = View.leftMembers(view, new_view);
                    List<Address> joinedMembers = View.newMembers(view, new_view);
                    if (view.getMembers().size() > new_view.getMembers().size()) {
                        if (!leftMembers.isEmpty()) {
                            raftListener.memberLeave(leftMembers);
                        }
                    } else {
                        if (!joinedMembers.isEmpty()) {
                            raftListener.memberJoin(joinedMembers);
                        }
                    }
                }
                view = new_view;
            }
        });
        this.jChannel.connect(DEFAULT_SERVER);
        logger.warn("server " + serverId + " started ...");
    }

    public void stop() throws Exception {
        if (this.jChannel != null) {
            this.jChannel.disconnect();
            this.jChannel.close();
        }
    }

    /****
     * 集群内推送信息
     * @param message
     * @throws Exception
     */
    public Object sendAll(Object message) throws Exception {
        this.jChannel.send(null, message);
        return message;
    }

    /***
     * 单节点发送消息
     * @param address
     * @param message
     * @return
     * @throws Exception
     */
    public Object send(Address address, Object message) throws Exception {
        this.jChannel.send(address, message);
        return message;
    }

    /***
     * 获取集群内所有节点地址
     *
     * @return
     */
    public List<Address> getServers() {
        return this.jChannel.getView().getMembers();
    }

    /***
     * 获取集群节点的leader
     * @return
     */
    public Address getLeader() {
        return this.raftHandle.raft().leader();
    }

    /***
     * 判断当前节点是否是leader
     * @return
     */
    public boolean isLeader() {
        return this.raftHandle.raft().isLeader();
    }

    /***
     * raft server接收消息的回调接口
     */
    interface RaftListener {

        default public void memberJoin(List<Address> address) {
            logger.warn("member join: " + address);
        }

        default public void memberLeave(List<Address> address) {
            logger.warn("member leave: " + address);
        }

        default public void receive(Message message) {
            logger.warn("receive message: " + message);
        }

        default public void isLeader() {
            logger.warn("current server is leader");
        }

        default public void isNotLeader() {

        }
    }
}
